/*------------------------------------------------------------------------------
* str2str.c : console version of stream server
*
*          Copyright (C) 2007-2020 by T.TAKASU, All rights reserved.
*
* version : $Revision: 1.1 $ $Date: 2008/07/17 21:54:53 $
* history : 2009/06/17  1.0  new
*           2011/05/29  1.1  add -f, -l and -x option
*           2011/11/29  1.2  fix bug on recognize ntrips:// (rtklib_2.4.1_p4)
*           2012/12/25  1.3  add format conversion functions
*                            add -msg, -opt and -sta options
*                            modify -p option
*           2013/01/25  1.4  fix bug on showing message
*           2014/02/21  1.5  ignore SIG_HUP
*           2014/08/10  1.5  fix bug on showing message
*           2014/08/26  1.6  support input format gw10, binex and rt17
*           2014/10/14  1.7  use stdin or stdout if option -in or -out omitted
*           2014/11/08  1.8  add option -a, -i and -o
*           2015/03/23  1.9  fix bug on parsing of command line options
*           2016/01/23  1.10 enable septentrio
*           2016/01/26  1.11 fix bug on station position by -p option (#126)
*                            add option -px
*           2016/07/01  1.12 support CMR/CMR+
*           2016/07/23  1.13 add option -c1 -c2 -c3 -c4
*           2016/09/03  1.14 support ntrip caster
*                            add option -ft,-fl
*           2016/09/06  1.15 add reload soure table by USR2 signal
*           2016/09/17  1.16 add option -b
*           2017/05/26  1.17 add input format tersus
*           2020/11/30  1.18 support api change strsvrstart(),strsvrstat()
*-----------------------------------------------------------------------------*/
#include <signal.h>
#include <stdbool.h>
#include <unistd.h>
#include "rtklib.h"

#define PRGNAME     "str2str"          /* program name */
#define MAXSTR      5                  /* max number of streams */
#define TRFILE      "str2str.trace"    /* trace file */

/* global variables ----------------------------------------------------------*/
static strsvr_t strsvr;                /* stream server */
static volatile int intrflg=0;         /* interrupt flag */

/* help text -----------------------------------------------------------------*/
static const char *help[]={
"",
" usage: nmea_player [-in stream] [-out stream [-out stream...]] [options]",
"",
"",
" -in  stream[#format] input  stream path and format",
" -out stream[#format] output stream path and format",
"",
"  stream path",
"    serial       : serial://port[:brate[:bsize[:parity[:stopb[:fctr]]]]]",
"    tcp server   : tcpsvr://:port",
"    tcp client   : tcpcli://addr[:port]",
"    ntrip client : ntrip://[user[:passwd]@]addr[:port][/mntpnt]",
"    ntrip server : ntrips://[:passwd@]addr[:port]/mntpnt[:str] (only out)",
"    ntrip caster : ntripc://[user:passwd@][:port]/mntpnt[:srctbl] (only out)",
"    file         : [file://]path[::T][::+start][::xseppd][::S=swap]",
"",
" -n  msec          nmea request cycle (m) [0]",
" -h                print help",

"./nmea_player.exe -in BBinary2022-11-09_145235.out -out tcpcli://127.0.0.1:8001 -in RBinary2022-11-09_145257.out -out tcpcli://127.0.0.1:8002 -n 100",
};
/* print help ----------------------------------------------------------------*/
static void printhelp(void)
{
    int i;
    for (i=0;i<sizeof(help)/sizeof(*help);i++) fprintf(stderr,"%s\n",help[i]);
    exit(0);
}
/* signal handler ------------------------------------------------------------*/
static void sigfunc(int sig)
{
    intrflg=1;
}
/* decode format -------------------------------------------------------------*/
static void decodefmt(char *path, int *fmt)
{
    char *p;
    
    *fmt=-1;
    
    if ((p=strrchr(path,'#'))) {
        if      (!strcmp(p,"#rtcm2")) *fmt=STRFMT_RTCM2;
        else if (!strcmp(p,"#rtcm3")) *fmt=STRFMT_RTCM3;
        else if (!strcmp(p,"#nov"  )) *fmt=STRFMT_OEM4;
        else if (!strcmp(p,"#oem3" )) *fmt=STRFMT_OEM3;
        else if (!strcmp(p,"#ubx"  )) *fmt=STRFMT_UBX;
        else if (!strcmp(p,"#ss2"  )) *fmt=STRFMT_SS2;
        else if (!strcmp(p,"#hemis")) *fmt=STRFMT_CRES;
        else if (!strcmp(p,"#stq"  )) *fmt=STRFMT_STQ;
        else if (!strcmp(p,"#javad")) *fmt=STRFMT_JAVAD;
        else if (!strcmp(p,"#nvs"  )) *fmt=STRFMT_NVS;
        else if (!strcmp(p,"#binex")) *fmt=STRFMT_BINEX;
        else if (!strcmp(p,"#rt17" )) *fmt=STRFMT_RT17;
        else if (!strcmp(p,"#sbf"  )) *fmt=STRFMT_SEPT;
        else return;
        *p='\0';
    }
}
/* decode stream path --------------------------------------------------------*/
static int decodepath(const char *path, int *type, char *strpath, int *fmt)
{
    char buff[1024],*p;
    
    strcpy(buff,path);
    
    /* decode format */
    decodefmt(buff,fmt);
    
    /* decode type */
    if (!(p=strstr(buff,"://"))) {
        strcpy(strpath,buff);
        *type=STR_FILE;
        return 1;
    }
    if      (!strncmp(path,"serial",6)) *type=STR_SERIAL;
    else if (!strncmp(path,"tcpsvr",6)) *type=STR_TCPSVR;
    else if (!strncmp(path,"tcpcli",6)) *type=STR_TCPCLI;
    else if (!strncmp(path,"ntripc",6)) *type=STR_NTRIPCAS;
    else if (!strncmp(path,"ntrips",6)) *type=STR_NTRIPSVR;
    else if (!strncmp(path,"ntrip", 5)) *type=STR_NTRIPCLI;
    else if (!strncmp(path,"file",  4)) *type=STR_FILE;
    else {
        fprintf(stderr,"stream path error: %s\n",buff);
        return 0;
    }
    strcpy(strpath,p+3);
    return 1;
}
/* read receiver commands ----------------------------------------------------*/
static void readcmd(const char *file, char *cmd, int type)
{
    FILE *fp;
    char buff[MAXSTR],*p=cmd;
    int i=0;
    
    *p='\0';
    
    if (!(fp=fopen(file,"r"))) return;
    
    while (fgets(buff,sizeof(buff),fp)) {
        if (*buff=='@') i++;
        else if (i==type&&p+strlen(buff)+1<cmd+MAXRCVCMD) {
            p+=sprintf(p,"%s",buff);
        }
    }
    fclose(fp);
}

struct play_t
{
    char *pin;
    int type_in;
    int fmt_in;
    char *pout;
    int type_out;
    int fmt_out;
    int ms;
    int n;
    int max;
    gtime_t sol_time;
};

struct play_t list[MAXSTR];
/*
pthread_mutex_t mutex = PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
bool test_cond = false;
*/
void stream_sync(gtime_t t, int ms,int n)
{
    list[n].sol_time = t;
    int i = 0;

    /*全部解出时间则继续*/
/*    while(1)
    {
        for(i = 0; i < list[n].max; i++)
        {
            if(list[i].sol_time.time == 0)
            {
                break;
            }
        }

        if(i == list[n].max)
        {
            break;
        }
        usleep(ms*10);
    }
*/
    /*如果有时间滞后的就停下来等待*/
    int diff = 0;
    int sync_ok = 1;
    while(1)
    {
        for(i = 0; i < list[n].max; i++)
        {
            if(list[i].sol_time.time < t.time)
            {
                break;
            }

            if(diff < (list[i].sol_time.time - t.time))
            {
                diff = (list[i].sol_time.time - t.time);
            }
        }

        if(i == list[n].max)
        {
            if(diff > 10)/*如果落后超过十秒则减小延时时间*/
            {
                sync_ok = 0;
            }
            break;
        }
        usleep(ms*10);
    }

    if(sync_ok == 1)
    {
        printf("sync:%d,%d\n",n,ms);
        usleep(ms*1000);
    }
    else
    {
        printf("quick sync:%d\n",n);
        usleep(ms*10);
    }
}

int one_stream(char *pin, int type_in, int fmt_in, char *pout, int type_out, int fmt_out, int ms, int n)
{
    stream_t strin, strout;
    int ret = 0;
    int rn = 0;
    int i;
    char buf[16];
    gtime_t time={0};
    solbuf_t solbuf={0};
    initsolbuf(&solbuf, 1, 2);

    strinit(&strin);
    strinit(&strout);

    ret = stropen(&strin, type_in, fmt_in, pin);
    if(ret != 1)
    {
        printf("open input failed\n");
        exit(-1);
    }
    ret = stropen(&strout, type_out, fmt_out, pout);
    if(ret != 1)
    {
        printf("open output failed:%s\n", pout);
        exit(-1);
    }

    while(1)
    {
        /*printf("read\n");*/
        rn = strread(&strin, buf, sizeof(buf));

        for(i = 0; i < rn; i++)
        {
            ret = inputsol(buf[i], time,time, 0.0, 0, &solopt_default,&solbuf);
            if(ret == 1)
            {
                sol_t *sol = getsol(&solbuf, 0);
                printf("%s\n", time_str(sol->time, 2));
                stream_sync(sol->time, ms, n);
            }
        }

        strwrite(&strout, buf, rn);

        if(rn == 0)
        {
            printf("finish\n");
            break;
        }
    }

    freesolbuf(&solbuf);
    exit(0);
}

void thread_stream(void *arg)
{
    struct play_t *list = (struct play_t *)arg;
    one_stream(list->pin, list->type_in, list->fmt_in, list->pout, list->type_out, list->fmt_out, list->ms, list->n);
}


/* str2str -------------------------------------------------------------------*/
int main(int argc, char **argv)
{
    static char cmd_strs[MAXSTR][MAXRCVCMD]={"","","","",""};
    static char cmd_periodic_strs[MAXSTR][MAXRCVCMD]={"","","","",""};
    const char ss[]={'E','-','W','C','C'};
    strconv_t *conv[MAXSTR]={NULL};
    double pos[3],stapos[3]={0},stadel[3]={0};
    static char s1[MAXSTR][MAXSTRPATH]={{0}},s2[MAXSTR][MAXSTRPATH]={{0}};
    static char s3[MAXSTR][MAXSTRPATH]={{0}};
    char *paths[MAXSTR],*logs[MAXSTR];
    char *cmdfile[MAXSTR]={"","","","",""},*cmds[MAXSTR],*cmds_periodic[MAXSTR];
    char *local="",*proxy="",*msg="1004,1019",*opt="",buff[256],*p;
    char strmsg[MAXSTRMSG]="",*antinfo="",*rcvinfo="";
    char *ant[]={"","",""},*rcv[]={"","",""},*logfile="";
    int i,j,n=0,dispint=5000,trlevel=0,opts[]={10000,10000,2000,32768,10,1000,30,0};
    int types[MAXSTR]={STR_FILE,STR_FILE},stat[MAXSTR]={0},log_stat[MAXSTR]={0};
    
    int byte[MAXSTR]={0},bps[MAXSTR]={0},fmts[MAXSTR]={0},sta=0;
    
    char *paths_in[MAXSTR];
    int types_in[MAXSTR]={STR_FILE,STR_FILE};
    int fmts_in[MAXSTR]={0};
    int n_in = 0;

    int pn = 0;

    pthread_t thread;
    pthread_attr_t attr;
    pthread_attr_init(&attr);

    memset(list, 0, sizeof(list));
    for (i=0;i<MAXSTR;i++) {
        paths[i]=s1[i];
        paths_in[i]=s3[i];
        logs[i]=s2[i];
        cmds[i]=cmd_strs[i];
        cmds_periodic[i]=cmd_periodic_strs[i];
    }
    for (i=1;i<argc;i++) {
        if (!strcmp(argv[i],"-in")&&i+1<argc) {
            if (!decodepath(argv[++i],types_in + n_in,paths_in[n_in],fmts_in+n_in)) return -1;
            n_in++;
        }
        else if (!strcmp(argv[i],"-out")&&i+1<argc&&n<MAXSTR-1) {
            if (!decodepath(argv[++i],types+n,paths[n],fmts+n)) return -1;
            printf("%s\n", paths[0]);
            n++;
        }
        else if (!strcmp(argv[i],"-p")&&i+3<argc) {
            pos[0]=atof(argv[++i])*D2R;
            pos[1]=atof(argv[++i])*D2R;
            pos[2]=atof(argv[++i]);
            pos2ecef(pos,stapos);
        }
        else if (!strcmp(argv[i],"-px")&&i+3<argc) {
            stapos[0]=atof(argv[++i]);
            stapos[1]=atof(argv[++i]);
            stapos[2]=atof(argv[++i]);
        }
        else if (!strcmp(argv[i],"-o")&&i+3<argc) {
            stadel[0]=atof(argv[++i]);
            stadel[1]=atof(argv[++i]);
            stadel[2]=atof(argv[++i]);
        }
        else if (!strcmp(argv[i],"-msg")&&i+1<argc) msg=argv[++i];
        else if (!strcmp(argv[i],"-opt")&&i+1<argc) opt=argv[++i];
        else if (!strcmp(argv[i],"-sta")&&i+1<argc) sta=atoi(argv[++i]);
        else if (!strcmp(argv[i],"-d"  )&&i+1<argc) dispint=atoi(argv[++i]);
        else if (!strcmp(argv[i],"-s"  )&&i+1<argc) opts[0]=atoi(argv[++i]);
        else if (!strcmp(argv[i],"-r"  )&&i+1<argc) opts[1]=atoi(argv[++i]);
        else if (!strcmp(argv[i],"-n"  )&&i+1<argc) opts[5]=atoi(argv[++i]);
        else if (!strcmp(argv[i],"-f"  )&&i+1<argc) opts[6]=atoi(argv[++i]);
        else if (!strcmp(argv[i],"-c"  )&&i+1<argc) cmdfile[0]=argv[++i];
        else if (!strcmp(argv[i],"-c1" )&&i+1<argc) cmdfile[1]=argv[++i];
        else if (!strcmp(argv[i],"-c2" )&&i+1<argc) cmdfile[2]=argv[++i];
        else if (!strcmp(argv[i],"-c3" )&&i+1<argc) cmdfile[3]=argv[++i];
        else if (!strcmp(argv[i],"-c4" )&&i+1<argc) cmdfile[4]=argv[++i];
        else if (!strcmp(argv[i],"-a"  )&&i+1<argc) antinfo=argv[++i];
        else if (!strcmp(argv[i],"-i"  )&&i+1<argc) rcvinfo=argv[++i];
        else if (!strcmp(argv[i],"-l"  )&&i+1<argc) local=argv[++i];
        else if (!strcmp(argv[i],"-x"  )&&i+1<argc) proxy=argv[++i];
        else if (!strcmp(argv[i],"-b"  )&&i+1<argc) opts[7]=atoi(argv[++i]);
        else if (!strcmp(argv[i],"-fl" )&&i+1<argc) logfile=argv[++i];
        else if (!strcmp(argv[i],"-t"  )&&i+1<argc) trlevel=atoi(argv[++i]);
        else if (*argv[i]=='-') printhelp();
    }
    if (n<=0) n=1; /* stdout */

    for(pn = 0; pn < n; pn++)
    {
        list[pn].pin = paths_in[pn];
        list[pn].type_in = types_in[pn];
        list[pn].fmt_in = fmts_in[pn];
        list[pn].pout = paths[pn];
        list[pn].type_out = types[pn];
        list[pn].fmt_out = fmts[pn];
        list[pn].ms = opts[5];
        list[pn].n = pn;
        list[pn].max = n;
        printf("%d\n", pn);
        pthread_create(&thread, &attr, thread_stream, &list[pn]);
    }
    while(1)
    {
        usleep(1000*1000);
    }
 /*   one_stream(paths_in[0], types_in[0], fmts_in[0], paths[0], types[0], fmts[0], opts[5]);*/

    for (i=0;i<n;i++) {
        if (fmts[i+1]<=0) continue;
        if (fmts[i+1]!=STRFMT_RTCM3) {
            fprintf(stderr,"unsupported output format\n");
            return -1;
        }
        if (fmts[0]<0) {
            fprintf(stderr,"specify input format\n");
            return -1;
        }
        if (!(conv[i]=strconvnew(fmts[0],fmts[i+1],msg,sta,sta!=0,opt))) {
            fprintf(stderr,"stream conversion error\n");
            return -1;
        }
        strcpy(buff,antinfo);
        for (p=strtok(buff,","),j=0;p&&j<3;p=strtok(NULL,",")) ant[j++]=p;
        strcpy(conv[i]->out.sta.antdes,ant[0]);
        strcpy(conv[i]->out.sta.antsno,ant[1]);
        conv[i]->out.sta.antsetup=atoi(ant[2]);
        strcpy(buff,rcvinfo);
        for (p=strtok(buff,","),j=0;p&&j<3;p=strtok(NULL,",")) rcv[j++]=p;
        strcpy(conv[i]->out.sta.rectype,rcv[0]);
        strcpy(conv[i]->out.sta.recver ,rcv[1]);
        strcpy(conv[i]->out.sta.recsno ,rcv[2]);
        matcpy(conv[i]->out.sta.pos,stapos,3,1);
        matcpy(conv[i]->out.sta.del,stadel,3,1);
    }
    signal(SIGTERM,sigfunc);
    signal(SIGINT ,sigfunc);
    signal(SIGHUP ,SIG_IGN);
    signal(SIGPIPE,SIG_IGN);
    
    strsvrinit(&strsvr,n+1);
    
    if (trlevel>0) {
        traceopen(*logfile?logfile:TRFILE);
        tracelevel(trlevel);
    }
    fprintf(stderr,"stream server start\n");
    
    strsetdir(local);
    strsetproxy(proxy);
    
    for (i=0;i<MAXSTR;i++) {
        if (*cmdfile[i]) readcmd(cmdfile[i],cmds[i],0);
        if (*cmdfile[i]) readcmd(cmdfile[i],cmds_periodic[i],2);
    }
    /* start stream server */
    if (!strsvrstart(&strsvr,opts,types,paths,logs,conv,cmds,cmds_periodic,
                     stapos)) {
        fprintf(stderr,"stream server start error\n");
        return -1;
    }
    for (intrflg=0;!intrflg;) {
        
        /* get stream server status */
        strsvrstat(&strsvr,stat,log_stat,byte,bps,strmsg);
        
        /* show stream server status */
        for (i=0,p=buff;i<MAXSTR;i++) p+=sprintf(p,"%c",ss[stat[i]+1]);
        
        fprintf(stderr,"%s [%s] %10d B %7d bps %s\n",
                time_str(utc2gpst(timeget()),0),buff,byte[0],bps[0],strmsg);
        
        sleepms(dispint);
    }
    for (i=0;i<MAXSTR;i++) {
        if (*cmdfile[i]) readcmd(cmdfile[i],cmds[i],1);
    }
    /* stop stream server */
    strsvrstop(&strsvr,cmds);
    
    for (i=0;i<n;i++) {
        strconvfree(conv[i]);
    }
    if (trlevel>0) {
        traceclose();
    }
    fprintf(stderr,"stream server stop\n");
    return 0;
}
